package reviews.task1;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class ProductScoreReducer extends Reducer<Text, IntWritable, Text, DoubleWritable> {
    /**
     * 存放聚合对象的列表
     */
    List<ProductScore> list = new ArrayList<ProductScore>();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        // 1. 定义一个计数器
        int count = 0;
        int sum_score = 0;
        // 2. 遍历一组迭代器，把每一个数量1累加起来
        for (IntWritable value : values) {
            count += 1;
            sum_score += value.get();
        }
        double avg_score = count == 0 ? 0.0 : sum_score / count;
        // 3. 将聚合后的结果保存到一个ProductScore对象当中
        ProductScore productScore = new ProductScore(key.toString(), avg_score);
        // 4. 将对象添加到List集合当中
        list.add(productScore);
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // 1. 对集合进行排序
        Collections.sort(list);
        // 2. 定义一个计数器
        int num = 0;
        // 3. 遍历集合
        for (ProductScore productScore : list) {
            // 4. 输出最终的结果
            context.write(new Text(productScore.getProductId()), new DoubleWritable(productScore.getScoreValue()));
            // 5. 计数器加1
            num++;
            // 6. 判断计数器是否等于10
            if (num == 20) {
                // 7. 当计数器等于10时退出
                return;
            }
        }
    }
}
